from redis.asyncio import Redis
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
from app.settings.config import settings
from app.log.log import logger


async def get_redis_connection():
    redis_url = f"redis://{settings.REDIS['username']}:{settings.REDIS['password']}@{settings.REDIS['host']}:{settings.REDIS['port']}/{settings.REDIS['db']}"
    redis = await Redis.from_url(redis_url, encoding="utf-8", decode_responses=True)
    return redis

async def init_kafka():
    # 启动Kafka生产者
    producer = AIOKafkaProducer(
        bootstrap_servers=f"{settings.MQ['host']}:{settings.MQ['port']}",
        value_serializer=lambda v: v.encode('utf-8'),
    )
    await producer.start()
    
    # 启动Kafka消费者（后台任务）
    consumer = AIOKafkaConsumer(
        settings.MQ['consumer']['topics'],
        bootstrap_servers=f"{settings.MQ['host']}:{settings.MQ['port']}",
        group_id=settings.MQ['consumer']['group_id'],
        value_deserializer=lambda v: v.decode('utf-8'),
    )
    await consumer.start()
    
    return producer, consumer